package com.fwmagic.spark.streaming

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object UpdateStateByKeyStreaming {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf()
                .setAppName(this.getClass.getSimpleName)
                .setMaster("local[*]")

        //默认会创建SparkContext[StreamingContext.createNewSparkContext(conf)]
        val ssc = new StreamingContext(conf, Seconds(5))

        //设置日志级别
        ssc.sparkContext.setLogLevel("WARN")

        /**
          * 对历史数据进行处理，累加历史批次数据
          * 累加数据就要将中间结果保存起来【要求保存在靠谱的存储系统中【安全】】，这样以后任务失败可以从State中恢复数据
          */
        //设置checkpoint保存状态数据
        ssc.checkpoint("file:///Users/fangwei/learn/mycode/workspace/fwmagic-spark/ch")

        //一系列Transformation操作
        val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 8888)

        val wordAndOne: DStream[(String, Int)] = lines.flatMap(_.split("\\s")).map((_,1))

        /**
          * updateFunc: (Seq[V], Option[S]) => Option[S]
          * 第一个参数：当前批次的数据
          * 第二个参数：初始值或者中间累加结果
          */
        /*
        //定义函数
        val updateFunc = (s: Seq[Int], op: Option[Int]) => {
            //将当前批次累加结果和初始值或者中间累加值进行累加
            Some(s.sum + op.getOrElse(0))
        }
        */
        //定义方法
        def updateFunc(s: Seq[Int], op: Option[Int]):Option[Int] ={
            Some(s.sum + op.getOrElse(0))
        }

        //有状态的累加
        val result: DStream[(String, Int)] = wordAndOne.updateStateByKey(updateFunc)

        //Action
        result.print()

        //开启任务
        ssc.start()

        //等待任务停止
        ssc.awaitTermination()
    }


}
